package com.qqs.stock.handler;

import com.qqs.stock.constant.DeductStockConstant;
import com.qqs.stock.dao.DeductStockDao;
import com.qqs.stock.dao.LocalDeductStockDao;
import com.qqs.stock.dto.DeductStockOrderPromise;
import com.qqs.stock.exception.UniqueException;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class DefaultDeductStockMerge extends Thread implements DeductStockMerge {
    private int capacity = 10;

    private final Thread handler = new Thread(this::handler);
    private final BlockingQueue<DeductStockOrderPromise> queue = new LinkedBlockingQueue<>(capacity);

    private DeductStockDao deductStockDao = new LocalDeductStockDao();

    public DefaultDeductStockMerge() {

    }

    public DefaultDeductStockMerge(int capacity) {
        this.capacity = capacity;
    }

    @Override
    public void run() {
        handler.run();
    }

    @Override
    public boolean merge(DeductStockOrderPromise promise) throws InterruptedException {
        return merge(promise, 100, TimeUnit.SECONDS);
    }

    @Override
    public boolean merge(DeductStockOrderPromise promise, long timeout, TimeUnit unit) throws InterruptedException {
        return queue.offer(promise, timeout, unit);
    }

    @Override
    public void handler() {
        while (true) {
            if (queue.isEmpty()) {
                try {
                    Thread.sleep(10);
                    continue;
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            // 取出任务
            List<DeductStockOrderPromise> promises = new ArrayList<>();
            int size = queue.size();
            for (int i = 0; i < size; i++) {
                promises.add(queue.poll());
            }

            boolean downgrade = false;
            try {
                // 处理任务之前先上锁
                promises.forEach(DeductStockOrderPromise::lock);
                int sum = promises.stream().mapToInt(DeductStockOrderPromise::getCount).sum();
                final Long warId = promises.get(0).getWarId();
                final Long orderId = promises.get(0).getOrderId();

                int dbSum = deductStockDao.selectCountByWareId(warId);
                if (dbSum > sum) {
                    // 开启事务
                    deductStockDao.openTransaction();
                    try {
                        // 插入流水
                        for (DeductStockOrderPromise promise : promises) {
                            // 插入流水可能会出现异常,比如：幂等性导致同一订单尝试扣减多次,
                            // 上个请求在当前请求执行到这里时才提交事务或已插入未提交事务导致唯一索引异常
                            try {
                                deductStockDao.insertBill(orderId, warId, promise.getCount());
                            } catch (UniqueException e) {
                                e.printStackTrace();
                                // 调整要扣减的库存数量
                                sum = sum - promise.getCount();
                                fillOrSignal(promise, DeductStockConstant.Status.INSERT_BILL_ERROR, true);
                            }
                        }
                        // 再次查询库存?执行到这里就没必要再查询了,前面的查询是为了保证有库存后才插入流水
                        downgrade = !deductStockDao.deduct(warId, sum);
                        if (!downgrade) {
                            // 没有触发降级则提交事务
                            deductStockDao.submitTransaction();
                            fillOrSignal(promises, DeductStockConstant.Status.SUCCESS, true);
                        }
                    } catch (Exception e) {
                        // 未知异常,直接回滚事务
                        e.printStackTrace();
                        deductStockDao.rollbackTransaction();
                        fillOrSignal(promises, DeductStockConstant.Status.UNKNOWN_ERROR, true);
                    }
                } else {
                    downgrade = true;
                }

                // 库存不足,进行降级处理
                if (downgrade) {
                    // 触发降级,先回滚流水
                    deductStockDao.rollbackTransaction();
                    // 从大到小进行尝试
                    List<DeductStockOrderPromise> sortedPromise = promises.stream()
                            .sorted((p1, p2) -> p2.getCount() - p1.getCount()).collect(Collectors.toList());

                    for (DeductStockOrderPromise promise : sortedPromise) {
                        // 开启事务
                        deductStockDao.openTransaction();
                        // 插入流水
                        try {
                            deductStockDao.insertBill(orderId, warId, promise.getCount());
                        } catch (UniqueException e) {
                            e.printStackTrace();
                            // 回滚事务
                            deductStockDao.rollbackTransaction();
                            fillOrSignal(promise, DeductStockConstant.Status.INSERT_BILL_ERROR, true);
                            // next
                            continue;
                        } catch (Exception e) {
                            e.printStackTrace();
                            // 回滚事务
                            deductStockDao.rollbackTransaction();
                            fillOrSignal(promise, DeductStockConstant.Status.UNKNOWN_ERROR, true);
                            // next
                            continue;
                        }
                        // 尝试扣减库存
                        boolean deduct = deductStockDao.deduct(warId, promise.getCount());
                        if (deduct) {
                            // 事务提交
                            deductStockDao.submitTransaction();
                            fillOrSignal(promise, DeductStockConstant.Status.SUCCESS, true);
                        } else {
                            // 回滚事务
                            deductStockDao.rollbackTransaction();
                            fillOrSignal(promise, DeductStockConstant.Status.NOT_ENOUGH_STOCK, true);
                        }
                    }
                }
            } finally {
                promises.forEach(DeductStockOrderPromise::lock);
            }
        }
    }

    private void fillOrSignal(List<DeductStockOrderPromise> promises, DeductStockConstant.Status status, boolean signal) {
        for (DeductStockOrderPromise promise : promises) {
            promise.setStatus(status.getCode());
            promise.setMessage(status.getMessage());
            if (signal) {
                promise.signal();
            }
        }
    }

    private void fillOrSignal(DeductStockOrderPromise promise, DeductStockConstant.Status status, boolean signal) {
        promise.setStatus(status.getCode());
        promise.setMessage(status.getMessage());
        if (signal) {
            promise.signal();
        }
    }
}
